package homework2.miningPopularSong.reducer;

import homework2.miningPopularSong.program.MiningPopularSong;

import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class ReduceRound1 extends Reducer<Text, Text, Text, Text>{

	@Override
	protected void cleanup(Context context) throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		super.cleanup(context);
	}

	@Override
	protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {

		String[] inputCountries = context.getConfiguration().getStrings(MiningPopularSong.LCOUNTRY);
		String[] country=new String[inputCountries.length];
		for(int i=0;i<inputCountries.length;i++){
			country[i]="";
		}
		Map<String, Integer> countSongsForUserid = new HashMap<>();
		int count = 0;
		
		for (Text mappedValue : values) {
			/* if the value belong to the "MapFileUserProfileRound1" */
			if (mappedValue.toString().startsWith("$$")) {
				if (count != inputCountries.length) {
					for(int i=0;i<inputCountries.length;i++){
						if (inputCountries[i].contains(mappedValue.toString().substring(2)) &&
								country[i].compareTo("") == 0) {
							country[i] = mappedValue.toString();
							count++;
						}
						else {
							continue;
						}
					}	
				}
			} 
			/* if the value belong to the "MapFileSongsListenedRound1" */
			else {
				if (countSongsForUserid.containsKey(mappedValue.toString())) {
					countSongsForUserid.put(mappedValue.toString(), countSongsForUserid.get(mappedValue.toString()) + 1);
				} else {
					countSongsForUserid.put(mappedValue.toString(), 1);
				}
			}
		}
		
		for(int i=0;i<inputCountries.length;i++){
			if (country[i].compareTo("") != 0 && countSongsForUserid.size() > 0) {
				Iterator<Entry<String, Integer>> iterator = countSongsForUserid.entrySet().iterator();
				
				/* for each song played */
				while (iterator.hasNext()) {
					Entry<String, Integer> song = iterator.next();
					context.write(key, new Text(country[i] + "\t" + song.getKey() + "\t" + song.getValue()));
					iterator.remove();
				}
			}
		}	
	}

	@Override
	public void run(Context arg0) throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		super.run(arg0);
	}

	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		super.setup(context);
	}
	
}
